/**
 * 
 */
package config;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.util.List;

import mapper.GraphMapper;

import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputLogFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import partitioner.PartitionDAO;
import partitioner.PartitionDAODatabaseImpl;
import reducer.GraphReducer;
import util.ArrayUtil;
import util.DBUtil;

/**
 * @author puneet
 * 
 */
public class ShortestPathJobRunnerWithDAO {

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Option sourceOption = new Option("s", "source", true, "Source Node");
		Option destOption = new Option("d", "dest", true, "Source Node");
		Options options = new Options();
		options.addOption(sourceOption);
		options.addOption(destOption);

		Configuration conf = new Configuration();
		
		GenericOptionsParser optionsParser = new GenericOptionsParser(conf,
				options, args);
		
		String source = optionsParser.getCommandLine().getOptionValue("s");
		String dest = optionsParser.getCommandLine().getOptionValue("d");
		System.out.println("Source = " + source + " Dest=" + dest);
		Connection conn = null;
		try {
			conn = DBUtil.getConnection();
			PartitionDAO dao = new PartitionDAODatabaseImpl(conn);
			FileSystem fs = FileSystem.get(conf);
			if (!dao.containsNode(source) || !dao.containsNode(dest)) {
				throw new Exception(
						"Source or Destnation does not exist in the i/p file");
			}
			Path tempMapFilePath = new Path("tmp/mapinput");
			if (fs.exists(tempMapFilePath)) {
				fs.delete(tempMapFilePath, false);
			}
			FSDataOutputStream out = fs.create(tempMapFilePath);
			String tempFileContents = source + " 0 "
					+ ArrayUtil.join(dao.getNeighbors(source), " ") + "\n";
			System.out.println("Writing to Map:" + tempFileContents);
			out.write(tempFileContents.getBytes());
			out.close();
			Configuration jobConfig = new Configuration();
			Job job = new Job(jobConfig, "graph search map reduce");
			job.setMapperClass(GraphMapper.class);
			job.setReducerClass(GraphReducer.class);
			job.setJarByClass(ShortestPathJobRunnerWithDAO.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(IntWritable.class);

			MyInputFormat.addInputPath(job, tempMapFilePath);

			Path tempOutputDir = new Path("tmp/op/");
			if (fs.exists(tempOutputDir)) {
				fs.delete(tempOutputDir, true);
			}
			FileOutputFormat.setOutputPath(job, tempOutputDir);
			job.waitForCompletion(true);
			int i = 0;
			while (true) {
				System.out.println("Running for:" + (++i));
				Path[] outputFileList = FileUtil.stat2Paths(fs
						.listStatus(tempOutputDir, new OutputLogFilter()));

				tempMapFilePath = new Path("tmp/mapinput");
				if (fs.exists(tempMapFilePath)) {
					fs
							.copyToLocalFile(tempMapFilePath, new Path(
									"lastmapinput"));
					fs.delete(tempMapFilePath, false);
				}
				out = fs.create(tempMapFilePath);
				for (Path opfile : outputFileList) {
					BufferedReader br = new BufferedReader(
							new InputStreamReader(fs.open(opfile)));
					String s = null;
					while ((s = br.readLine()) != null) {
						// System.out.println("Line = " + s);
						String node = s.split("\\s")[0];
						if (node == null || "".equals(node)) {
							continue;
						}
						// System.out.println("Doing Node = "+node);
						if (node.equals(dest)) {
							System.out.println("Shortest Distance Found:"
									+ s.split("\\s")[1]);
							out.close();
							return;
						}
						out.write(s.getBytes());
						List<String> neighbors = dao.getNeighbors(node);
						if(neighbors == null) {
							System.out.println("Neighbor of "+node+ " are null");
						}else{
							for (String neighbor : neighbors) {
								out.write((" " + neighbor).getBytes());
							}
							out.write("\n".getBytes());
						}
						
						// toWrite.append(ArrayUtil.join(dao.getNeighbors(s.split("\\s")[0]),"\t"));

						// System.out.println("Writing to map:"+toWrite);
						// out.writeChars(toWrite.toString());

						System.gc();
					}
				}
				out.close();
				fs.delete(tempOutputDir, true);
				jobConfig = new Configuration();
				job = new Job(jobConfig, "graph search map reduce");
				job.setMapperClass(GraphMapper.class);

				job.setReducerClass(GraphReducer.class);
				job.setJarByClass(ShortestPathJobRunner.class);
				job.setOutputKeyClass(Text.class);
				job.setOutputValueClass(IntWritable.class);

				MyInputFormat.addInputPath(job, tempMapFilePath);
				tempOutputDir = new Path("tmp/op/");
				if (fs.exists(tempOutputDir)) {
					fs.delete(tempOutputDir, true);
				}
				FileOutputFormat.setOutputPath(job, tempOutputDir);
				job.waitForCompletion(true);
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			DBUtil.closeConnection(conn);
		}
		// Now make the first file
		// Configuration config = new Configuration();
		// Job job = new Job(config, "graph map reduce");
		// job.setJarByClass(JobRunner.class);
		// job.setMapperClass(GraphMapper.class);
		// job.setReducerClass(GraphReducer.class);
		// job.setOutputKeyClass(Text.class);
		// job.setOutputValueClass(IntWritable.class);
		// TextInputFormat.addInputPath(job, new Path(args[0]));
		// FileOutputFormat.setOutputPath(job, new Path(args[1]));
		// System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}